package com.hylanda.nifi_determines_field;

import java.io.BufferedOutputStream;
import java.io.OutputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;

import org.apache.commons.io.IOUtils;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;

import com.alibaba.fastjson.JSONObject;
import com.hylanda.nifi_determines_field.util.GenUrlCrc64;
import com.hylanda.nifi_determines_field.util.NifiUtil;



@Tags({"字段判断  field json"})
@CapabilityDescription("Determines whether a field is included in json.")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class DeterminesFieldProcessor extends AbstractProcessor{

	public static final PropertyDescriptor FILED = new PropertyDescriptor
            .Builder().name("FILED")
            .displayName("FILED")
            .description("FILED")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final Relationship SUCCESS = new Relationship.Builder()
            .name("SUCCESS")
            .description("SUCCESS")
            .build();
    public static final Relationship FAILURE = new Relationship.Builder()
            .name("FAILURE")
            .description("FAILURE")
            .build();
    

    private List<PropertyDescriptor> descriptors;

    private Set<Relationship> relationships;

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
        descriptors.add(FILED);
        this.descriptors = Collections.unmodifiableList(descriptors);

        final Set<Relationship> relationships = new HashSet<Relationship>();
        relationships.add(SUCCESS);
        relationships.add(FAILURE);
        this.relationships = Collections.unmodifiableSet(relationships);
    }

    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {

    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        FlowFile flowFile = session.get();
        if ( flowFile == null ) {
            return;
        }
        ComponentLog log = getLogger();
        String field = context.getProperty(FILED).getValue();
        String content = NifiUtil.getContent(session, flowFile);
        
        if (null==content) {
        	log.info("content is null");
        	session.transfer(flowFile, FAILURE);
			return;
		}
        JSONObject jsonObject = JSONObject.parseObject(content);
        if (null == jsonObject) {
        	log.info("jsonObject is null");
        	session.transfer(flowFile, FAILURE);
			return;
		}
        
        
        flowFile = session.write(flowFile, (in, out) -> {
			try (OutputStream outputStream = new BufferedOutputStream(out)) {				
				outputStream.write(IOUtils.toByteArray(content));
			}
		});
        
        
        String uuid = GenUrlCrc64.GenCrc64(UUID.randomUUID().toString());
        String session_id = flowFile.getAttribute(CoreAttributes.UUID.key());
        String instance_id = getIdentifier();
        JSONObject logJSON = new JSONObject();
        JSONObject msgJSON = new JSONObject();
        
        Date date = new Date();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String time = sdf.format(date);
        msgJSON.put("time", time);
        msgJSON.put("status", "success");
        
        logJSON.put("msg", msgJSON);
        logJSON.put("session_id", session_id);
        logJSON.put("uuid", uuid);
        logJSON.put("instance_id", instance_id);
        
        getLogger().info(logJSON.toJSONString());
        if (jsonObject.containsKey(field)) {
        	String value = jsonObject.getString(field);
        	if (null==value||value.equals("")) {
        		session.transfer(flowFile,FAILURE);
			}
        	else{
        		session.transfer(flowFile, SUCCESS);
        	}
		}else{
			session.transfer(flowFile,FAILURE);
		}
    }

}
